package mc.support.zmq.route;

import mc.support.zmq.err.ZmqTimeoutException;
import org.zeromq.ZMQ;

/**
 * Created with IntelliJ IDEA. <br/>
 * Author: Francis Yuen    <br/>
 * Date: 2014-10-27  <br/>
 */
public class ZmqConnection {

    private static final long DEFAULT_TIMEOUT = 5 * 1000;

    private final ZMQ.Socket socket;
    private final String bizType;
    private final String serverAddress;
    private TimeoutHandler timeoutHandler;

    public ZmqConnection( ZMQ.Socket socket, String bizType, String serverAddress ) {
        this.socket = socket;
        this.bizType = bizType;
        this.serverAddress = serverAddress;
    }

    public byte[] sendAndWaitRsp( byte[] bs ) {
        return sendAndWaitRsp( bs, DEFAULT_TIMEOUT );
    }

    public byte[] sendAndWaitRsp( byte[] bs, final long timeout ) {
        socket.send( bs, 0 );
        if ( ZMQ.poll( new ZMQ.PollItem[] {new ZMQ.PollItem( socket, ZMQ.Poller.POLLIN )}, timeout ) > 0 ) {
            return socket.recv( 0 );
        } else {
            if ( null != timeoutHandler ) timeoutHandler.handle( this, timeout );
            throw new ZmqTimeoutException( "no reply from " + serverAddress + " after " + timeout + " ms" );
        }
    }

    public void close() {
        socket.close();
    }

    public String getBizType() {
        return bizType;
    }

    public ZMQ.Socket getSocket() {
        return socket;
    }

    public String getServerAddress() {
        return serverAddress;
    }

    ZmqConnection setTimeoutHandler( TimeoutHandler timeoutHandler ) {
        this.timeoutHandler = timeoutHandler;
        return this;
    }

}
